#include <functional>
#include <thread>
#include <iostream>

#include "threadpool.h"

const int TASK_MAX_THRESHHOLD = 1024;
const int THREAD_MAX_THRESHHOLD = 100;
const int THREAD_MAX_IDLE_TIME = 10; // 单位: 秒

// 线程池构造
ThreadPool::ThreadPool()
    : initThreadSize_(0)
    , taskSize_(0)
    , idleThreadSize_(0)
    , curThreadSize_(0)
    , taskQueMaxThreshHold_(TASK_MAX_THRESHHOLD)
    , threadSizeThreshHold_(THREAD_MAX_THRESHHOLD)
    , poolMode_(PoolMode::MODE_FIXED)
    , isPoolRunning_(false)
{}

// 线程池析构
/*
* 假设所以线程都没事情做,都在notEmpty_等待,需要去唤醒,所以析构函数需要把它们都唤醒,
* 所以线程都从等待状态变为阻塞状态,然后去抢锁;
* 如果threads_.size() != 0, 就把锁释放掉等在条件变量上面, 等到全部的线程对象结束;
*/
ThreadPool::~ThreadPool()
{
    isPoolRunning_ = false;
    notEmpty_.notify_all(); // 发现死锁的位置

    // 等待线程池里面所有的线程返回 有两种状态: 阻塞 & 正在执行任务中
    std::unique_lock<std::mutex> lock(taskQueMtx_);
    //notEmpty_.notify_all();
    exitCond_.wait(lock, [&]()->bool {return threads_.size() == 0; });
}

// 设置线程池的工作模式
void ThreadPool::setMode(PoolMode mode)
{
    if (checkRunningState())
        return;
    poolMode_ = mode;
}

// 设置task任务队列上限阈值
void ThreadPool::setTaskQueMaxThreshHold(int threshhold)
{
    if (checkRunningState())
        return;
    taskQueMaxThreshHold_ = threshhold;
}

// 设置线程池cached模式下线程阈值
void ThreadPool::setThreadSizeThreshHold(int threshhold)
{
    if (checkRunningState())
        return;
    if (poolMode_ == PoolMode::NODE_CACHED)
    {
        threadSizeThreshHold_ = threshhold;
    }
}

// 给线程池提交任务  用户调用该接口, 传入任务对象, 生产任务
Result ThreadPool::submitTask(std::shared_ptr<Task> sp)
{
    // 获取锁 (构造函数自动调用lock: _Pmtx->lock() )
    std::unique_lock<std::mutex> lock(taskQueMtx_);

    // 线程的通信 等待任务队列有空余
    // 用户提交任务, 最长不能阻塞超过1s, 否则判断提交任务失败, 返回
    /*
    while(taskQue_.size() < taskQueMaxThreshHold_){
        notFull_.wait(lock);
    }
    */
    // 等价于上面的写法
    // wait():一直等  wait_for():等多长时间  wait_until():等到什么时候
    // wait_for()返回的是bool值,当在提交任务的时候,任务队列满了,在1s之内没有空位置,则提交任务失败,返回;
    if (!notFull_.wait_for(lock, std::chrono::seconds(1),
        [&]() {return taskQue_.size() < (size_t)taskQueMaxThreshHold_; }))
    {
        // 表示notFull_等待1s,条件依然没有满足
        std::cerr << "task queue is full, submit task fail." << std::endl;

        // 线程执行完task, task对象就被析构掉了; 所以Result需要用指针接收任务, 不然任务没有得到返回值就没了
        // 任务对象返回一个无效的返回值
        return Result(sp, false);
    }

    // 如果有空余, 把任务放入任务队列中
    taskQue_.emplace(sp);
    taskSize_++;

    // 因为新放了任务, 任务队列肯定不空了, 在notEmpty_上进行通知, 分配线程执行任务
    notEmpty_.notify_all();

    /*
    * cached模式: 任务处理比较紧急; 场景: 小而快的任务 需要根据任务数量和空闲线程的数量, 判断是否需要创建新的线程出来;
    * 
    * cached模式支持小而快的任务, 如果耗时的任务多了, 不适合cached模式, 因为耗时的任务会长时间占用一个线程,
    * 如果耗时任务比较多得话, 会导致线程池创建的线程过多; 如果整个系统线程创建的过多的话, 对系统的性能影响非常大,
    * 所以耗时的任务不宜用cached模式;
    */
    if (poolMode_ == PoolMode::NODE_CACHED
        && taskSize_ > idleThreadSize_
        && curThreadSize_ < threadSizeThreshHold_)
    {
        std::cout << ">>> create new thread... " << std::endl;

        // 创建新线程
        auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
        int threadId = ptr->getId();
        threads_.emplace(threadId, std::move(ptr));
        // 启动线程
        threads_[threadId]->start();
        // 修改线程个数相关的变量 
        curThreadSize_++;
        idleThreadSize_++;
    }

    // 返回任务的Result对象
    return Result(sp);
}

// 开启线程池
void ThreadPool::start(int initThreadSize)
{
    // 设置线程池的运行状态
    isPoolRunning_ = true;

    // 记录初始线程个数
    initThreadSize_ = initThreadSize;
    curThreadSize_ = initThreadSize;

    // 创建线程对象 std::vector<Thread*> threads_;
    for (int i = 0; i < initThreadSize_; i++)
    {
        // 创建thread线程对象的时候, 把线程函数给到thread线程对象 (summary.txt -> 8)
        auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));

        // unique_ptr不允许左值引用的拷贝构造函数和赋值运算符
        //           只允许右值引用的拷贝构造函数和赋值运算符
        //threads_.emplace_back(std::move(ptr));

        int threadId = ptr->getId();
        threads_.emplace(threadId, std::move(ptr));
    }

    // 启动所有线程 
    for (int i = 0; i < initThreadSize_; i++)
    {
        threads_[i]->start();  // 需要去执行一个线程函数
        idleThreadSize_++; // 记录初始空闲线程的数量(刚开始启动的线程肯定是空闲线程, 没有去执行任务)
    }
}

// 定义线程函数  线程池的所有线程从任务队列里面消费任务
void ThreadPool::threadFunc(int threadid)  // 线程函数返回, 相应的线程也就结束了
{
    // 线程执行任务的初始时间
    auto lastTime = std::chrono::high_resolution_clock().now();

    while (isPoolRunning_) 
    {
        std::shared_ptr<Task> task;
        {
            // 先获取锁
            std::unique_lock<std::mutex> lock(taskQueMtx_);

            std::cout << "tid:" << std::this_thread::get_id()
                << "尝试获取任务..." << std::endl;

            // cached模式下, 有可能已经创建了很多的线程, 但是空闲时间超过60s, 应该把多余的线程
            // 结束回收掉(超过initThreadSize_数量的线程要进行回收)
            // 当前时间 - 上一次线程执行的时间 > 60s
           
            // 每一秒中返回一次 怎么区分: 超时返回? 还是有任务待执行返回
            while (taskQue_.size() == 0) // 出现死锁之前的判断条件

            // 锁 + 双重判断
            //while (isPoolRunning_ && taskQue_.size() == 0)
            {
                if (poolMode_ == PoolMode::NODE_CACHED)
                {
                    // 每个1s都会检查一下有没有超过60s空闲的
                    // 条件变量, 超时返回了 (timeout超时了, no_timeout没超时)
                    if (std::cv_status::timeout == notEmpty_.wait_for(lock, std::chrono::seconds(1)))
                    {
                        auto now = std::chrono::high_resolution_clock().now();
                        // 时间差转为 秒 为单位; 
                        auto dur = std::chrono::duration_cast<std::chrono::seconds>(now - lastTime);
                        if (dur.count() >= THREAD_MAX_IDLE_TIME
                            && curThreadSize_ > initThreadSize_)
                        {
                            // 开始回收当前线程
                            // 记录线程数量的相关变量的值修改
                            // 把线程对象从线程列表容器中删除  办法: threadFunc <==> thread对象
                            // threadid ==> thread对象 ==> 删除
                            threads_.erase(threadid); // std::this_thread::get_id() 
                            curThreadSize_--;
                            idleThreadSize_--;

                            std::cout << "threadid: " << std::this_thread::get_id() << " exit!" << std::endl;
                            return; // 线程函数返回, 相应的线程也就结束了
                        }
                    }
                }
                else 
                {
                    // 等待notEmpty条件
                    notEmpty_.wait(lock);
                }

                if (!isPoolRunning_)
                {
                    threads_.erase(threadid); // std::this_thread::get_id() 
                    std::cout << "threadid: " << std::this_thread::get_id() << " exit!" << std::endl;
                    exitCond_.notify_all();
                    return; // 结束线程函数, 就是结束当前线程了
                }
            }

            idleThreadSize_--;

            std::cout << "tid:" << std::this_thread::get_id()
                << "获取任务成功..." << std::endl;

            // 从任务队列中取一个任务出来
            task = taskQue_.front();
            taskQue_.pop();
            taskSize_--;

            // 如果依然有剩余任务, 继续通知其他的线程执行任务
            if (taskQue_.size() > 0)
            {
                notEmpty_.notify_all();
            }

        } // 出作用域把锁释放掉,让其他线程再来获取锁,拿取任务

        // 当前线程负责执行这个任务
        if (task != nullptr) 
        {
            task->exec();
        }

        idleThreadSize_++;
        lastTime = std::chrono::high_resolution_clock().now();  // 更新线程执行完任务的时间
    }

    // 在这里回收资源是因为线程正在执行任务, 没有阻塞在条件变量上, 所以在这里回收
    threads_.erase(threadid); // std::this_thread::get_id() 
    std::cout << "threadid: " << std::this_thread::get_id() << " exit!" << std::endl;
    exitCond_.notify_all();
}

bool ThreadPool::checkRunningState() const
{
    return isPoolRunning_;
}


/******************** 线程方法实现 ********************/
int Thread::generateId_ = 0;

// 线程构造
Thread::Thread(ThreadFunc func)
    : func_(func)
    , threadId_(generateId_++)
{}

// 线程析构
Thread::~Thread()
{}

// 启动线程
void Thread::start()
{
    // 创建一个线程来执行一个线程函数
    std::thread t(func_, threadId_);  // 给线程函数传入线程ID
    t.detach();  // 设置分离线程 (原因:summary.txt -> 9)
}

int Thread::getId() const
{
    return threadId_;
}

/******************** Task方法的实现 ********************/

Task::Task()
    : result_(nullptr)
{}

void Task::exec()
{
    if (result_ != nullptr)
    {
        result_->setVal(run());  // 这里发生多态调用
    }
}

void Task::setResult(Result* res)
{
    result_ = res;
}

/******************** Result方法的实现 ********************/

Result::Result(std::shared_ptr<Task> task, bool isValid)
    : task_(task)
    , isValid_(isValid)
{
    task_->setResult(this);
}

Any Result::get()
{
    if (!isValid_)
    {
        return "";
    }

    sem_.wait(); // task任务如果没有执行完, 这里会阻塞用户的线程
    return std::move(any_);
}

void Result::setVal(Any any)
{
    // 存储task的返回值
    // unique_ptr没有左值引用的拷贝构造函数和赋值运算符
    this->any_ = std::move(any);
    sem_.post(); // 已经获取的任务的返回值, 增加信号量资源
}





